package main

import (
	"context"
	"crypto/tls"
	"encoding/json"
	"github.com/icetech233/ekuiper-kafka/conf"
	kafkago "github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl/scram"
	"time"
)

func tt_main22() {
	conf.Log.Info("test1 info")
	conf.Log.Debug("test1 debug")

	topic := "com/1720306450484363265/ruledata"
	partition := 0

	// "127.0.0.1:9092"
	// "123.60.49.33:9092"

	conn, err := kafkago.DialLeader(context.Background(), "tcp", "123.60.49.33:9092", topic, partition)
	if err != nil {
		conf.Log.Error("failed to dial leader:", err)
	}

	// 写入

	_, err = conn.WriteMessages(
		kafkago.Message{Value: rand(590.1, "2000833106", "2024-04-17 18:55:14")},
		kafkago.Message{Value: rand(595.4, "2000833106", "2024-04-17 18:55:14")},
		kafkago.Message{Value: rand(597.5, "2000833106", "2024-04-17 19:20:14")},
	)

	if err != nil {
		conf.Log.Fatal("failed to write messages:", err)
	}

	conf.Log.Debug("test1 debug")
	time.Sleep(time.Second * 5)
	conn.Close()
}

func main() {
	var err error
	conf.Log.Info("test1 info")

	topic := "com.1720306450484363265.ruledata"
	// partition := 0
	mechanism, _ := scram.Mechanism(scram.SHA512, "dGhhbmtmdWwtaGFkZG9jay02MjU4JAKmctzQBU_wDThdOZKFAg8-_SiZqxKk9hs",
		"ZDJmYmMyZDItYTE3Yy00NTNkLThhNjgtMmE5NzRlNmU4YWQ2")
	w := kafkago.Writer{
		Addr:  kafkago.TCP("thankful-haddock-6258-us1-kafka.upstash.io:9092"),
		Topic: topic,
		Transport: &kafkago.Transport{
			SASL: mechanism,
			TLS:  &tls.Config{},
		},
	}
	conf.Log.Info("test2 info")
	err = w.WriteMessages(context.Background(), kafkago.Message{Value: []byte("MESSAGE")})
	if err != nil {
		conf.Log.Fatal("write messages ", err)
	}
	conf.Log.Info("test3 info")
	err = w.Close()
	if err != nil {
		conf.Log.Fatal(err)
	}

}
func rand(cur_val, pd_motr_pnt_id, acg_tm any) []byte {
	m := map[string]any{
		"cur_val":        cur_val,
		"pd_motr_pnt_id": pd_motr_pnt_id,
		"acg_tm":         acg_tm,
	}
	var b []byte
	b, _ = json.Marshal(m)
	return b
}
